/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.broker.longpolling;

import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.ServiceThread;
import org.apache.rocketmq.common.SystemClock;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.store.ConsumeQueueExt;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class PullRequestHoldService extends ServiceThread {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
    private static final String TOPIC_QUEUEID_SEPARATOR = "@";
    private final BrokerController brokerController;
    private final SystemClock systemClock = new SystemClock();
    //pull请求缓存队列：当consumer发起pull请求时，broker中没有新消息会将pull请求缓存到此队列
    private ConcurrentMap<String/* topic@queueId */, ManyPullRequest> pullRequestTable =
        new ConcurrentHashMap<String, ManyPullRequest>(1024);

    public PullRequestHoldService(final BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    /**
     * 将需要Hold的pullRequest放入到一个map中，等待校验(consumer拉取消息时，broker中没有新的消息时会将请求hold住，等待轮询)
     *    没有消息时，被hold卡住的请求pullRequest，重新唤起的场景：被挂起请求定期扫描场景 + 新消息到来立即回复场景
     *          重新唤起的场景：没有消息，就卡住请求，轮询等待，每5秒重新查一次有没有新消息到了，如果有消息就会返回结果，如果发现请求过了15s，还没有消息，也会返回结果
     *          新消息到来立即回复场景： broker创建分区索引线程，在收到新消息，在创建topic分区索引后，会唤起topic分区对应的被挂起的请求
     *          
     *    具体校验代码在checkHoldRequest中
     *    
     * @param topic
     * @param queueId
     * @param pullRequest
     */
    public void suspendPullRequest(final String topic, final int queueId, final PullRequest pullRequest) {
        //本地缓存的 key=topic@queueId, value是pullRequest请求列表
        String key = this.buildKey(topic, queueId);
        ManyPullRequest mpr = this.pullRequestTable.get(key);
        if (null == mpr) {
            mpr = new ManyPullRequest();
            ManyPullRequest prev = this.pullRequestTable.putIfAbsent(key, mpr);
            if (prev != null) {
                mpr = prev;
            }
        }
        //将pullRequest加入到请求列表(因为同一个topic的一个分区，可能有多个消费组消费消息)
        mpr.addPullRequest(pullRequest);
    }

    private String buildKey(final String topic, final int queueId) {
        StringBuilder sb = new StringBuilder();
        sb.append(topic);
        sb.append(TOPIC_QUEUEID_SEPARATOR);
        sb.append(queueId);
        return sb.toString();
    }

    /**
     *  拉取消息pull请求缓存在broker端定期扫描场景：PullRequestHoldService线程会每隔5秒，遍历缓存在broker的pullRequest请求所查询的分区queue检测有没有新消息到来
     *      consumer发送拉取消息请求pullRequest ---》broker没有新的消息时，pullRequest会被暂时缓存到broker端的pullRequestTable中 
     *          ---》PullRequestHoldService线程每隔5秒检查broker是否有新消息到来  ---》有新消息到来，从缓存pullRequestTable中取出pullRequest响应给consumer端
     */
    @Override
    public void run() {
        log.info("{} service started", this.getServiceName());
        while (!this.isStopped()) {
            try {
                //如果开启长轮询每隔5秒判断消息是否到达（长轮询开启时，hold住的请求超时时间是15s）
                if (this.brokerController.getBrokerConfig().isLongPollingEnable()) {
                    this.waitForRunning(5 * 1000);
                } else {
                    //没有开启长轮询,每隔1s再次尝试(长轮询没开启时，hold住的请求超时时间是1s)
                    this.waitForRunning(this.brokerController.getBrokerConfig().getShortPollingTimeMills());
                }
                long beginLockTimestamp = this.systemClock.now();
                //检查是否有新消息到达broker中，有新消息到来就唤起拉取消息请求pullRequest，响应给consumer
                this.checkHoldRequest();
                long costTime = this.systemClock.now() - beginLockTimestamp;
                if (costTime > 5 * 1000) {
                    log.info("[NOTIFYME] check hold request cost {} ms.", costTime);
                }
            } catch (Throwable e) {
                log.warn(this.getServiceName() + " service has exception. ", e);
            }
        }

        log.info("{} service end", this.getServiceName());
    }

    @Override
    public String getServiceName() {
        return PullRequestHoldService.class.getSimpleName();
    }

    //不停的检查被hold住的请求：

    /**
     * 不停的检查broker是否有新消息到来 或 被hold住的请求超时：
     *      (1)检查是否有新消息到达broker中，有新消息到来就唤起hold住的拉取消息请求pullRequest，响应给消费者
     *      (2)检查在服务端hold住的pull请求是否超时，若超时则响应给消费者
     */
    private void checkHoldRequest() {
        for (String key : this.pullRequestTable.keySet()) {
            //解析key：key = topic@queueId
            String[] kArray = key.split(TOPIC_QUEUEID_SEPARATOR);
            if (2 == kArray.length) {
                String topic = kArray[0];
                int queueId = Integer.parseInt(kArray[1]);
                //获取指定messageQueue下最大的offset，然后跟当前的offset(pullRequest请求中的拉取消息的offset)比较，来确定是否有新消息到来
                final long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
                try {
                    //判断是否有新消息到来或超时，再将消息响应给发送pullRequest请求的consumer
                    this.notifyMessageArriving(topic, queueId, offset);
                } catch (Throwable e) {
                    log.error("check hold request failed. topic={}, queueId={}", topic, queueId, e);
                }
            }
        }
    }

    public void notifyMessageArriving(final String topic, final int queueId, final long maxOffset) {
        notifyMessageArriving(topic, queueId, maxOffset, null, 0, null, null);
    }

    /**
     *  缓存到broker中的pullRequest请求响应：
     *   
     *  当pull请求拉取消息时，broker没有新消息，会将pullRequest请求缓存到broker一段时间，在两种场景下会唤起挂起的请求，并响应给consumer。
     *  此方案触发的两个场景：
     *      1.主动检查：PullRequestHoldService#run()线程定期循环检查broker是否有新消息到来
     *      2.被动通知：长轮询开始时，若有新消息到来，会在创建索引的线程ReputMessageService#run中唤起hold住的请求
     *        
     * @param topic
     * @param queueId
     * @param maxOffset
     * @param tagsCode
     * @param msgStoreTime
     * @param filterBitMap
     * @param properties
     */
    public void notifyMessageArriving(final String topic, final int queueId, final long maxOffset, final Long tagsCode,
        long msgStoreTime, byte[] filterBitMap, Map<String, String> properties) {
        String key = this.buildKey(topic, queueId);
        ManyPullRequest mpr = this.pullRequestTable.get(key);
        if (mpr != null) {
            //获取同一个topic同一个queue分区下多个pull请求列表，并清空queueId下的请求
            List<PullRequest> requestList = mpr.cloneListAndClear();
            if (requestList != null) {
                List<PullRequest> replayList = new ArrayList<PullRequest>();
                //遍历该分区下的请求
                for (PullRequest request : requestList) {
                    long newestOffset = maxOffset;
                    if (newestOffset <= request.getPullFromThisOffset()) {
                        //重新获取topic分区中的MaxOffset
                        newestOffset = this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId);
                    }

                    //1.broker中有新消息，响应consumer：topic分区队列索引中最大maxOffset 大于 消费者消费偏移量
                    if (newestOffset > request.getPullFromThisOffset()) {
                        //tag过滤：基于消息分区consumeQueue
                        boolean match = request.getMessageFilter().isMatchedByConsumeQueue(tagsCode,
                            new ConsumeQueueExt.CqExtUnit(tagsCode, msgStoreTime, filterBitMap));
                        // match by bit map, need eval again when properties is not null.
                        //sql表达式过滤消息：基于commitLog具体的消息
                        if (match && properties != null) {
                            match = request.getMessageFilter().isMatchedByCommitLog(null, properties);
                        }
                        if (match) {
                            try {
                                //如果拉取消息偏移大于请求偏移量,如果消息匹配调用executeRequestWhenWakeup处理消息
                                this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                    request.getRequestCommand());
                            } catch (Throwable e) {
                                log.error("execute request when wakeup failed.", e);
                            }
                            continue;
                        }
                    }
                    
                    //2.超时响应：比较当前时间和阻塞的时间，pull请求缓存在broker中超时，响应consumer(长轮询开启时，默认超时时间15s，长轮询没有开启默认是1s)
                    if (System.currentTimeMillis() >= (request.getSuspendTimestamp() + request.getTimeoutMillis())) {
                        try {
                            this.brokerController.getPullMessageProcessor().executeRequestWhenWakeup(request.getClientChannel(),
                                request.getRequestCommand());
                        } catch (Throwable e) {
                            log.error("execute request when wakeup failed.", e);
                        }
                        continue;
                    }
                    //pull请求未超时 + topic分区队列没有新消息，再将pull请求放回broker缓存中
                    replayList.add(request);
                }

                if (!replayList.isEmpty()) {
                    mpr.addPullRequest(replayList);
                }
            }
        }
    }
}
